package com.ruoyi.collector.service;


import com.ruoyi.common.utils.DateUtils;
import org.graylog2.syslog4j.server.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.net.SocketAddress;
import java.text.SimpleDateFormat;
import java.util.*;

/**
*@Description: TODO
*@CreateTime: 2022-12-31  15:54
*@Author: chaozi
*@Version: 1.0
*/
@Component
@Service
public class SyslogListener implements CommandLineRunner {
    private static final Logger log = LoggerFactory.getLogger(SyslogListener.class);
    private SyslogServerIF syslogServer;
    private final int listSize = 100;
    private final int timeOut = 1000;
    private List<Map<String,String >> syslogMsg = new ArrayList<>(listSize);
    private boolean syslogServerRunning = false;
    SimpleDateFormat formatter = new SimpleDateFormat(DateUtils.YYYY_MM_DD_HH_MM_SS);
//    FlinkKafkaConsumer<String> source = new FlinkKafkaConsumer<>("sever_log_to_flink_consumer", new SimpleStringSchema(), props);


    @Override
    public void run(String... args) throws Exception {
        log.info("开始监听syslog数据~");
        syslogServerRunning=true;
        final Date[] startDate = {new Date()};
        // 创建syslog服务端，并设置监听端口为514，使用UDP协议
        syslogServer = SyslogServer.getInstance("udp");
        SyslogServerConfigIF config = syslogServer.getConfig();
        config.setPort(514);
        config.addEventHandler(new SyslogServerSessionlessEventHandlerIF() {
            @Override
            public void event(SyslogServerIF syslogServerIF, SocketAddress socketAddress, SyslogServerEventIF syslogServerEventIF) {
                Map<String,String > map=new LinkedHashMap<>();
                try{
                    // syslog日志
                    String message = syslogServerEventIF.getMessage();
                    map.put("Address",socketAddress.toString().split("/")[1]);
                    map.put("CollectDate",formatter.format(syslogServerEventIF.getDate()));
                    map.put("Message",message);
                    syslogMsg.add(map);
                    //log.info("接收到{}日志：{},",socketAddress, syslogMsg);
                    long diff = new Date().getTime() - startDate[0].getTime();
                    int size = syslogMsg.size();
                    if (size > listSize || diff > timeOut) {
                        startDate[0].setTime(System.currentTimeMillis());
//                    KafkaUtils.batchSend(msgs, topicName);
                        log.info("发送日志：{},", syslogMsg);
                        syslogMsg.clear();
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }

            }

            @Override
            public void exception(SyslogServerIF syslogServerIF, SocketAddress socketAddress, Exception e) {

            }

            @Override
            public void initialize(SyslogServerIF syslogServerIF) {

            }

            @Override
            public void destroy(SyslogServerIF syslogServerIF) {

            }
        });

        // 启动syslog服务端
        syslogServer.run();

    }


    public boolean startSyslogServer() {

        if (!syslogServerRunning) {
            // 启动syslog
            syslogServer.run();
            syslogServerRunning = true;
            return true;
        } else {
            log.warn("Syslog server is already running");
            return false;
        }
    }
    public boolean stopSyslogServer() {
        if (syslogServerRunning) {
            // 停止syslog服务端
            syslogServer.shutdown();
            syslogServerRunning = false;
            return true;
        } else {
            log.warn("Syslog server is not running");
            return false;
        }
    }
    public boolean checkSyslogServer() {
       return syslogServerRunning;
    }
}
